package com.duowan.realtime.thrift;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.duowan.realtime.thirft.api.BloomFilterException;
import com.duowan.realtime.thirft.api.BloomFilterGroupQuery;
import com.duowan.realtime.thirft.api.BloomFilterService.Iface;
import com.duowan.realtime.thirft.api.Constants;
import com.duowan.realtime.thirft.api.DistinctData;
import com.duowan.realtime.thirft.api.DistinctRequest;

public class BloomFilterServiceImpl implements Iface {

	private ProxyBloomFilterServiceImpl proxyBFImpl;
	private static final Logger LOG = LoggerFactory.getLogger("bflog");
	
	public BloomFilterServiceImpl() {
		proxyBFImpl = new ProxyBloomFilterServiceImpl();
	}

	@Override
	public Map<String, Integer> notContainsCountAndAdd(String bloomfilterGroup, List<BloomFilterGroupQuery> bfGroupQuery) throws BloomFilterException, TException {
		Map<String, Integer> resultMap = new ConcurrentHashMap<String, Integer>(); 
		for (BloomFilterGroupQuery query : bfGroupQuery) {
			int count = proxyBFImpl.notContainsCountAndAdd(bloomfilterGroup, query.getPartition(), query.getGroup(), query.getValues());
			resultMap.put(query.getGroup(), count);
			LOG.info("BloomfilterQuery values size:" + query.getValues().size() + ",result:" + query.getGroup() + " add " + count + " into Bloomfilter pool");
		}
		return resultMap;
	}

	@Override
	public String ping() throws TException {
		return Constants.PING_RESPONSE;
	}

	@Override
	public Map<String, List<DistinctData>> notContainsAndMark(String bloomfilterGroup, List<DistinctRequest> bfDistinctRequest) throws BloomFilterException, TException {
		Map<String,List<DistinctData>> resultMap = new ConcurrentHashMap<String, List<DistinctData>>(); 
		for (DistinctRequest request : bfDistinctRequest) {
			List<DistinctData> count = proxyBFImpl.notContainsAndMark(bloomfilterGroup, request.getPartition(), request.getGroup(), request.getDistinctDatas());
			resultMap.put(request.getGroup(), count);
//			LOG.info("bfDistinctRequest values size:" + request.getDistinctDatasSize() + ",result:" + request.getGroup() + " add " + count  + " into Bloomfilter pool");
		}
		return resultMap;
	}


}
